package org.acme.cluster.service;

import lombok.Data;
import org.acme.cluster.po.SysSms;
import org.jgroups.Address;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;
import java.util.logging.Logger;

@Service
public class SysSmsDispatchServiceImpl extends ReceiverAdapter implements SysSmsDispatchService, CommandLineRunner, DisposableBean {

    // 重试短信ID列表
    private final Vector<String> retryIdList = new Vector<>();

    Logger logger = Logger.getLogger(SysSmsDispatchServiceImpl.class.getName());

    @Value("${sms.max-scheduled-thread-core-count:10}")
    private int maxScheduledThreadCoreCount;

    @Value("${sms.max-executor-thread-core-count:20}")
    private int maxExecutorThreadCoreCount;

    @Resource
    private SysSmsService sysSmsService;

    @Resource
    private SysClusterService sysClusterService;

    // 定时任务线程池
    private ScheduledExecutorService scheduledExecutorService;

    //执行线程池
    private ExecutorService executorService;

    // 集群视图
    private View view;

    //工作线程
    private Thread thread;

    //运行标识
    private volatile boolean running = true;

    /****
     * 执行任务
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        this.scheduledExecutorService = new ScheduledThreadPoolExecutor(this.maxScheduledThreadCoreCount);
        this.executorService = Executors.newFixedThreadPool(this.maxExecutorThreadCoreCount);
        this.thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (running) {
                    List<SysSms> list = sysSmsService.findByServerKey(sysClusterService.getUuid().toStringLong(), new ArrayList<>(retryIdList));
                    if (list.isEmpty()) {
                        LockSupport.park();
                    } else {
                        CountDownLatch countDownLatch = new CountDownLatch(list.size());
                        for (SysSms sysSms : list) {
                            executorService.execute(new SysSmsSendTask(sysSms.getId(), sysSmsService, countDownLatch));
                        }
                        try {
                            countDownLatch.await();
                        } catch (InterruptedException e) {
                            logger.warning("线程暂停失败: " + e.getMessage());
                        }
                    }
                }
            }
        });
        this.thread.setName("smsSendThread");
        this.thread.start();
    }

    /****
     * 程序结束时，释放资源
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        this.running = false;
        if (this.thread.isAlive()) {
            this.thread.interrupt();
        }
        this.executorService.shutdownNow();
        this.scheduledExecutorService.shutdownNow();
    }

    /***
     * 刷新任务，唤醒本节点线程
     * @return
     */
    @Override
    public boolean refreshTask() {
        LockSupport.unpark(thread);
        return true;
    }

    /****
     * 发送集群消息，唤醒集群节点的任务
     * @return
     * @throws Exception
     */
    @Override
    public boolean refreshALLTask() throws Exception {
        sysClusterService.sendAll("refreshTask");
        return true;
    }

    /****
     * 收到消息，那么就直接刷新
     * @param msg
     */
    @Override
    public void receive(Message msg) {
        this.refreshTask();
    }

    /****
     * 监听集群节点掉线了。
     * @param view
     */
    @Override
    public void viewAccepted(View view) {
        if (this.view != null) {
            List<Address> addresses = View.leftMembers(this.view, view);
            if (!addresses.isEmpty()) {
                try {
                    this.sysSmsService.updateSmsStatusToAll(view.getMembers(), sysClusterService.getUuid(), addresses);
                    this.refreshALLTask();
                } catch (Exception e) {
                    logger.warning("节点掉线，刷新任务失败: " + e.getMessage());
                }
            }
        }
        this.view = view;
    }

    @Override
    public View getView() {
        return view;
    }

    /****
     *
     * 短信發送線程
     */
    @Data
    class SysSmsSendTask implements Runnable {

        private String id;

        private SysSmsService sysSmsService;

        private CountDownLatch countDownLatch;

        public SysSmsSendTask(String id, SysSmsService sysSmsService, CountDownLatch countDownLatch) {
            this.id = id;
            this.sysSmsService = sysSmsService;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            SysSms sysSms = sysSmsService.getById(id);
            boolean flag = sysSmsService.sendSms(sysSms);
            if (!flag) {
                retryIdList.add(sysSms.getId());
                long time = sysSms.getNextRetryTime().getTime() - System.currentTimeMillis();
                scheduledExecutorService.schedule(new SysSmsReTryTask(sysSms.getId(), sysSmsService), time, TimeUnit.MILLISECONDS);
            }
            countDownLatch.countDown();
        }
    }

    @Data
    class SysSmsReTryTask implements Runnable {

        private String id;

        private SysSmsService sysSmsService;

        public SysSmsReTryTask(String id, SysSmsService sysSmsService) {
            this.id = id;
            this.sysSmsService = sysSmsService;
        }

        @Override
        public void run() {
            SysSms sysSms = sysSmsService.getById(id);
            if (sysSms != null) {
                boolean flag = sysSmsService.sendSms(sysSms);
                if (flag) {
                    retryIdList.remove(id);
                } else {
                    if (sysSms.getRetryCount() < sysSmsService.getMaxRetryCount()) {
                        long time = sysSms.getNextRetryTime().getTime() - System.currentTimeMillis();
                        if (!retryIdList.contains(id)) {
                            retryIdList.add(id);
                        }
                        scheduledExecutorService.schedule(new SysSmsReTryTask(sysSms.getId(), sysSmsService), time, TimeUnit.MILLISECONDS);
                    }
                }
            }
        }
    }
}
